package com.action.gulimall.seckill.service.impl;

import com.action.common.constant.MqConstant;
import com.action.common.constant.OrderConstant;
import com.action.common.to.mq.SeckillOrderTo;
import com.action.common.utils.R;
import com.action.common.vo.MemberResponseVo;
import com.action.gulimall.seckill.config.SeckillCommonConfig;
import com.action.gulimall.seckill.feign.ConpenFeignService;
import com.action.gulimall.seckill.feign.ProductFeignService;
import com.action.gulimall.seckill.interceptor.LoginInterceptor;
import com.action.gulimall.seckill.service.SeckillService;
import com.action.gulimall.seckill.to.SeckillSkuRedisTo;
import com.action.gulimall.seckill.vo.SeckillSessionWithSkusVo;
import com.action.gulimall.seckill.vo.SeckillSkuVo;
import com.action.gulimall.seckill.vo.SkuInfoVo;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RSemaphore;
import org.redisson.api.RedissonClient;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.BoundHashOperations;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

/**
 * @author sl shilun217@qq.com
 * @Deacription 商品秒杀业务具体实现
 * @date 2021/07/05/10:52
 **/
@Slf4j
@Service
public class SeckillServiceImpl implements SeckillService {

    @Autowired
    StringRedisTemplate redisTemplate;
    @Autowired
    RedissonClient redissonClient;
    @Autowired
    RabbitTemplate rabbitTemplate;
    @Autowired
    ConpenFeignService conpenFeignService;
    @Autowired
    ProductFeignService productFeignService;

    /**
     * 定时提前将需要秒杀的商品和库存放到redis缓存中进行上架，加快查询速度
     * 每天凌晨0点，查询未来三天的需要秒杀的场次和商品信息，放入redis
     * 幂等 Redisson分布式锁保证分布式情况下只有一个任务执行 key=seckill:upload:lock
     * <p>
     * 秒杀场次信息
     * key=seckill:sessions:startTime_endTime  value=[skuId1,skuId2...]
     * 秒杀商品信息
     * key=seckill:skus  key1=场次sessionId_skuId value1=商品JSON
     * 库存信号量   限流 商品需要有商品秒杀随机码 秒杀请求需要携带秒杀商品的随机码确保安全   商品库存量作为库存信号量数量，秒杀一次减一个信号量
     * key=seckill:stock:随机码token  value=库存量
     * 需要加过期时间
     * 存入缓存的时候需要判断特点key是否存在，存在就不需要再放入了，因为已经上架过了
     */
    @Override
    public void upSeckillSessionWithSkusFuture3Days() {
        //获取未来3天可以秒杀的场次和商品信息
        R r = conpenFeignService.getSeckillSessionWithSkusFuture3Days();
        if (r.getCode() != 0) {
            log.error("获取未来3天可以秒杀的场次和商品信息出现异常:" + JSON.toJSONString(r));
        } else {
            //数据解析
            List<SeckillSessionWithSkusVo> seckillSessionWithSkusVos = r.getData(new TypeReference<List<SeckillSessionWithSkusVo>>() {
            }).get();
            if (seckillSessionWithSkusVos != null) {
                //缓存秒杀数据
                seckillSessionWithSkusVos.forEach(s -> {
                    //缓存秒杀场次
                    upSeckillSession(s);
                    //缓存秒杀场次对应的商品和库存
                    upSeckillSkusAndStock(s);
                });
            }
        }
    }

    //缓存秒杀场次
    private void upSeckillSession(SeckillSessionWithSkusVo seckillSessionWithSkusVo) {
        //获取当前活动的开始和结束时间的时间戳
        long startTime = seckillSessionWithSkusVo.getStartTime().getTime();
        long endTime = seckillSessionWithSkusVo.getEndTime().getTime();
        String snTime = startTime + "_" + endTime;
        String seckillSessionKey = SeckillCommonConfig.SECKILL_SESSIONS + snTime;
        //判断Redis中是否有该信息，如果没有才进行添加
        Boolean hasKey = redisTemplate.hasKey(seckillSessionKey);
        if (!hasKey) {
            List<String> skuIds = seckillSessionWithSkusVo.getRelationSkus().stream()
                    .map(s -> s.getPromotionSessionId() + "_" + s.getSkuId())
                    .collect(Collectors.toList());
            redisTemplate.opsForList().leftPushAll(seckillSessionKey, skuIds);
        }
    }

    //缓存秒杀场次对应的商品和库存
    private void upSeckillSkusAndStock(SeckillSessionWithSkusVo seckillSessionWithSkusVo) {
        BoundHashOperations<String, Object, Object> boundHashOperations = redisTemplate.boundHashOps(SeckillCommonConfig.SECKILL_SKUS);
        List<SeckillSkuRedisTo> seckillSkuRedisTos = seckillSessionWithSkusVo.getRelationSkus().stream()
                .filter(s -> {
                    //过滤已经存在的
                    String key = seckillSessionWithSkusVo.getId().toString() + "_" + s.getSkuId().toString();
                    //判断Redis中是否有该信息，如果没有才进行添加
                    return !boundHashOperations.hasKey(key);
                }).map(s -> {
                    //构建秒杀商品缓存项
                    String token = UUID.randomUUID().toString().replace("_", "");
                    SeckillSkuRedisTo seckillSkuRedisTo = new SeckillSkuRedisTo();
                    //属性复制
                    BeanUtils.copyProperties(s, seckillSkuRedisTo);
                    //设置随机码
                    seckillSkuRedisTo.setRandomCode(token);
                    //设置当前商品的秒杀时间信息
                    seckillSkuRedisTo.setStartTime(seckillSessionWithSkusVo.getStartTime().getTime());
                    seckillSkuRedisTo.setEndTime(seckillSessionWithSkusVo.getEndTime().getTime());
                    return seckillSkuRedisTo;
                }).collect(Collectors.toList());

        if(CollectionUtils.isEmpty(seckillSkuRedisTos)){
           return;
        }
        List<Long> skuIds = seckillSkuRedisTos.stream().map(s -> s.getSkuId()).collect(Collectors.toList());
        //批量获取售卖商品数据
        R r = productFeignService.getSkuInfos(skuIds);
        if (r.getCode() != 0) {
            log.error("批量获取售卖商品数据出现异常:" + JSON.toJSONString(r));
        } else {
            List<SkuInfoVo> skuInfoVos = r.getData(new TypeReference<List<SkuInfoVo>>() {
            }).get();
            if(skuInfoVos!=null){
                //建立映射map
                Map<String, SkuInfoVo> skuInfoMap = skuInfoVos.stream().collect(Collectors.toMap(
                        s -> seckillSessionWithSkusVo.getId().toString() + "_" + s.getSkuId().toString(),
                        s -> s));
                //构建批量秒杀商品缓存map
                Map<String, String> seckillSkuRedisToMap = seckillSkuRedisTos.stream().collect(Collectors.toMap(
                        s -> s.getPromotionSessionId().toString() + "_" + s.getSkuId().toString(),
                        s -> {
                            String key = s.getPromotionSessionId().toString() + "_" + s.getSkuId().toString();
                            s.setSkuInfo(skuInfoMap.get(key));
                            return JSON.toJSONString(s);
                        }));
                //批量缓存秒杀商品
                boundHashOperations.putAll(seckillSkuRedisToMap);
                //构建秒杀商品库存信号量
                seckillSkuRedisTos.forEach(s -> {
                    String key =SeckillCommonConfig.SECKILL_STOCK_SEMAPHORE+s.getRandomCode();
                    // 使用库存作为分布式信号量
                    RSemaphore semaphore = redissonClient.getSemaphore(key);
                    // 商品可以秒杀的数量作为信号量
                    semaphore.trySetPermits(s.getSeckillCount());
                });
            }
        }
    }

    @Override
    public List<SeckillSkuRedisTo> getCurrentSeckillSkus() {
        List<SeckillSkuRedisTo> seckillSkuRedisTos = new ArrayList<SeckillSkuRedisTo>();
        Set<String> keys = redisTemplate.keys(SeckillCommonConfig.SECKILL_SESSIONS+"*");
        //seckill:sessions:startTime_endTime
         keys.stream().filter(key -> {
            String[] split = key.split("\\:")[2].split("_");
            long startTime = Long.valueOf(split[0]).longValue();
            long endTime = Long.valueOf(split[1]).longValue();
            long currentTime = System.currentTimeMillis();
            if (currentTime > startTime && currentTime < endTime) {
                return true;
            } else {
                return false;
            }
        }).forEach(key -> {
            List<String> seckillSessionSkuIds = redisTemplate.opsForList().range(key, 0, 100);
            BoundHashOperations<String, String, String> boundHashOperations = redisTemplate.boundHashOps(SeckillCommonConfig.SECKILL_SKUS);
            assert seckillSessionSkuIds != null;
            List<String> seckillSkuRedisToJsons = boundHashOperations.multiGet(seckillSessionSkuIds);
            if (seckillSkuRedisToJsons != null){
                List<SeckillSkuRedisTo> seckillSkuRedisTosTemp = seckillSkuRedisToJsons.stream().map(s -> JSON.parseObject(s, SeckillSkuRedisTo.class)).collect(Collectors.toList());
                seckillSkuRedisTos.addAll(seckillSkuRedisTosTemp);
            }
        });
        return seckillSkuRedisTos;
    }

    @Override
    public SeckillSkuVo getSeckillInfoBySkuId(Long skuId) {
        SeckillSkuVo seckillSkuVo = new SeckillSkuVo();
        //1、找到所有需要秒杀的商品的key信息---seckill:skus
        BoundHashOperations<String, String, String> hashOps = redisTemplate.boundHashOps(SeckillCommonConfig.SECKILL_SKUS);
        Set<String> keys = hashOps.keys();
        if(keys!=null&&keys.size()>0){
            for (String key : keys) {
                String reg = "\\d_"+skuId.toString();
                if(Pattern.matches(reg, key)){
                    String seckillSkuRedisToJson = hashOps.get(key);
                    SeckillSkuRedisTo seckillSkuRedisTo = JSON.parseObject(seckillSkuRedisToJson,SeckillSkuRedisTo.class);
                    //随机码
                    Long currentTime = System.currentTimeMillis();
                    Long startTime = seckillSkuRedisTo.getStartTime();
                    Long endTime = seckillSkuRedisTo.getEndTime();
                    //如果当前时间大于等于秒杀活动开始时间并且要小于活动结束时间
                    if (currentTime >= startTime && currentTime <= endTime) {
                        BeanUtils.copyProperties(seckillSkuRedisTo, seckillSkuVo);
                        return seckillSkuVo;
                    }
                    seckillSkuRedisTo.setRandomCode(null);
                    BeanUtils.copyProperties(seckillSkuRedisTo, seckillSkuVo);
                    return seckillSkuVo;
                }
            }
        }
        return null;
    }

    //秒杀业务
    @Override
    public String kill(String killId, String key, Integer num) {
        /**
        *    获取redis商品信息
         *   验证(秒杀时间判断，sessionId_skuId对应的随机码对比,秒杀数量限制，是否已经秒杀过)
         *   尝试信号量减扣库存
         *   快速下单，用户 orderId 秒杀场次 skuId 秒杀数量 秒杀价格  组装消息发送到MQ
         *   订单项目需要提前设置好队列和交换机绑定关系和秒杀消息监听和创单逻辑
        * */
        long now = System.currentTimeMillis();
        //获取当前用户的信息
        MemberResponseVo user = LoginInterceptor.threadLocal.get();
        BoundHashOperations<String, String, String> hashOperations = redisTemplate.boundHashOps(SeckillCommonConfig.SECKILL_SKUS);
        String seckillSkuRedisToJson = hashOperations.get(killId);
        //验证秒杀商品是否存在
        if(!StringUtils.isEmpty(seckillSkuRedisToJson)){
            SeckillSkuRedisTo seckillSkuRedisTo = JSON.parseObject(seckillSkuRedisToJson, SeckillSkuRedisTo.class);
            Long startTime = seckillSkuRedisTo.getStartTime();
            Long endTime = seckillSkuRedisTo.getEndTime();
            //验证秒杀商品是否过期
            if(now>=startTime&&now<=endTime){
                String token = seckillSkuRedisTo.getRandomCode();
                //验证随机码
                if(key.equals(token)){
                    //验证秒杀数量限制
                    if(num<=seckillSkuRedisTo.getSeckillLimit()){
                        String[] split = killId.split("_");
                        String seesionId = split[0];
                        String skuId = split[1];
                        //验证是否已经秒杀过了
                        String killedKey = user.getId() + "-" + skuId;
                        //设置自动过期(活动结束时间-当前时间)  需要保证原子性
                        Long ttl = endTime - now;
                        Boolean unKilled = redisTemplate.opsForValue().setIfAbsent(killedKey, num.toString(), ttl, TimeUnit.MICROSECONDS);
                        if(unKilled){
                            //验证信号量是否充足
                            RSemaphore semaphore = redissonClient.getSemaphore(SeckillCommonConfig.SECKILL_STOCK_SEMAPHORE + token);
                            if(semaphore.tryAcquire(num)){
                                //构建业务消息
                                String orderSn = redisTemplate.opsForList().rightPop(OrderConstant.ORDERSN_GENERATE_KEY);
                                if(StringUtils.isEmpty(orderSn)){
                                    orderSn = IdWorker.get32UUID();
                                }
                                // 用户id orderSn
                                SeckillOrderTo seckillOrderTo = new SeckillOrderTo();
                                seckillOrderTo.setOrderSn(orderSn);
                                seckillOrderTo.setMemberId(user.getId());
                                seckillOrderTo.setNum(num);
                                seckillOrderTo.setPromotionSessionId(seckillSkuRedisTo.getPromotionSessionId());
                                seckillOrderTo.setSkuId(seckillSkuRedisTo.getSkuId());
                                seckillOrderTo.setSeckillPrice(seckillSkuRedisTo.getSeckillPrice());
                                seckillOrderTo.setCreateTime(now);
                                rabbitTemplate.convertAndSend(MqConstant.ORDER_EXCHANGE_NAME,
                                        MqConstant.SECKILL_ORDER_TO_CREATE_ROUTINGKEY,
                                        seckillOrderTo, new CorrelationData(orderSn));
                                return orderSn;
                            }
                        }
                    }
                }
            }
        }
        return null;
    }
}
